package com.learn.lb.storm.wordcount;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author laibo
 * @since 2019/9/18 09:39
 */
public class WcTopology {

    private static final Logger log = LoggerFactory.getLogger(WcTopology.class);

    public static void main(String[] args) throws InterruptedException {
        TopologyBuilder builder = new TopologyBuilder();
        //spout并发数
        int spoutParal = 1;
        //创建spout，不能含有空格
        builder.setSpout("wcSpout", new WcSpout(), spoutParal);
        //指定bolt的并发数
        int boltParal = 1;
        //创建Bolt，localOrShuffleGrouping指定接受的spout产出的数据，并且以shuffle方式（即每个spout随机轮询发送tuple到下一个Bolt中）
        builder.setBolt("wcBolt", new WcBolt(), boltParal)
                .localOrShuffleGrouping("wcSpout");
        Config config = new Config();
        //表示整个topology将使用几个worker
        config.put(Config.TOPOLOGY_WORKERS, 2);
        //创建一个本地模式cluster
        LocalCluster cluster = new LocalCluster();
        //提交topology
        cluster.submitTopology("WcTopology", config, builder.createTopology());
        //等待1分钟， 1分钟后会停止拓扑和集群， 视调试情况可增大该数值
        TimeUnit.MINUTES.sleep(1);
        log.info("=========> 结束拓扑");
        cluster.killTopology("WcTopology");
        cluster.shutdown();

        //提交topology任务到集群上
//        StormSubmitter.submitTopology("WcTopology", config, builder.createTopology());
    }
}
